package com.lagou.rpc.consumer.client;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.consumer.handler.RpcClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.security.SecureRandom;
import java.util.*;
import java.util.concurrent.*;

/**
 * Rpc客户端
 */
public class RpcClient {
    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    private String ip;

    private int port;

    private RpcClientHandler rpcClientHandler = new RpcClientHandler();
    private ExecutorService executorService = Executors.newCachedThreadPool();

    /**
     * 定时线程池
     */
    private static ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(5);

    /**
     * Zookeeper客户端
     */
    private ZkClient zkClient = new ZkClient("0.0.0.0:2181");

    /**
     * 已连接服务端列表
     */
    private static List<String> connectedServerList = new ArrayList<>();

    /**
     * 服务端地址与Channe的集合
     */
    private static Map<String,Channel> channelMap = new HashMap<>();

    private static boolean init = true;

    private static String nameSpace = "/server/";

    public RpcClient(String ip,int port){
        this.ip = ip;
        this.port = port;
        initRpcClient();
    }

    public RpcClient() {
        initRpcClient();
    }


    /**
     * 初始化Rpc客户端
     */
    private void initRpcClient() {
        Bootstrap bootstrap = new Bootstrap();
        //连接Rpc服务端
        try {
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,2000)
                    .option(ChannelOption.SO_KEEPALIVE,Boolean.TRUE)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(rpcClientHandler);
                        }
                    });
            //连接服务端
            connect(bootstrap);
            if(init) {
                scheduledLog();
                init = false;
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            this.close();
        }

    }

    /**
     * 与服务端建立连接，监听zk服务端节点
     */
    public void connect(Bootstrap bootstrap ) throws InterruptedException {
        // 获取/Server下注册的所有服务端
        List<String> servers = zkClient.getChildren("/server");
        for (String server : servers) {
            // 客户端下线
            if(!zkClient.exists(nameSpace + server + "/status")) {
                continue;
            }
            // 服务端已建立连接，不再重新建立连接
            if(connectedServerList.contains(server)) {
                continue;
            }
            //server ip:port
            String[] data = server.split(":");
            // 获取/Server下注册的所有服务端
            Channel channel = bootstrap.connect(data[0], Integer.parseInt(data[1])).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if(channelFuture.isSuccess()) {

                        connectedServerList.add(server);
                        System.out.println("服务端已连接上线，路径： "+ server);

                        //连接客户端后，监听服务端节点
                        zkClient.subscribeChildChanges(nameSpace + server, new IZkChildListener() {
                            @Override
                            public void handleChildChange(String s, List<String> list) throws Exception {
                                if(CollectionUtils.isEmpty(list)){
                                    Channel offChannel = channelMap.remove(server);
                                    offChannel.close();
                                    connectedServerList.remove(server);//从服务端列表中移除
                                    System.out.println("服务端已下线，ip："+data[0]+",端口："+data[1]);
                                }else {
                                    //服务端上线，重新连接
                                    connect(bootstrap);
                                }
                            }
                        });
                    }
                }
            }).channel();
            channelMap.put(server,channel);
            System.out.println("客户端连接服务端成功,ip: "+data[0]+",port: "+data[1]);
        }
    }

    /**
     * 定时记录请求耗时以及系统时间
     */
    public void scheduledLog(){
        scheduledPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                List<String> serverNodes = zkClient.getChildren("/server");
                for (String serverNode : serverNodes) {
                    String data = zkClient.readData(nameSpace+serverNode);
                    if(StringUtils.isEmpty(data)) {
                        continue;
                    }
                   //判断服务端节点状态，是否下线
                    if(!zkClient.exists(nameSpace + serverNode + "/status")) {
                        continue;
                    }
                    Map nodeData = JSON.parseObject(data, Map.class);
                    long lastRequest = (long) nodeData.get("lastReqTime");//耗时
                    if(System.currentTimeMillis() - lastRequest > 5000) {
                        System.out.println("当前服务端响应时间失效，服务端地址：" + serverNode);
                        nodeData.put("ttlb",0);
                        zkClient.writeData(nameSpace + serverNode,JSON.toJSONString(nodeData));
                    }
                }
            }
        },5,5,TimeUnit.SECONDS);
    }

    /**
     * 资源关闭
     */
    public void close(){
        zkClient.close();
        if(!CollectionUtils.isEmpty(channelMap)) {
            for (Channel channel : channelMap.values()) {
                channel.close();
            }
        }
        if(null != eventLoopGroup) {
            eventLoopGroup.shutdownGracefully();
        }
    }

    /**
     * 消息发送方法
     * @return
     */
    public Object sendMsg(String msg) throws ExecutionException, InterruptedException {
        //负载均衡策略
        String serverPath = null;
        int currentTTLB = Integer.MAX_VALUE;
        List<String> serverNodes = zkClient.getChildren("/server");
        for (String serverNode : serverNodes) {
            String data = zkClient.readData(nameSpace + serverNode);
            if(StringUtils.isEmpty(data)) {
                serverPath = serverNode;
                break;
            }
            //判断服务端节点状态，是否下线
            if(!zkClient.exists(nameSpace + serverNode + "/status")) {
                continue;
            }
            Map nodeData = JSON.parseObject(data, Map.class);
            System.out.println("当前节点信息：" + JSON.toJSONString(nodeData));
            int ttlb = (int) nodeData.get("ttlb");//耗时
            if(ttlb < currentTTLB) {
                currentTTLB = ttlb;
                serverPath = serverNode;
            }
        }

        // 没有获取到最短耗时服务器，随机获取
        if(serverPath == null) {
            //随机获取服务端
            int serverIndex = new SecureRandom().nextInt(connectedServerList.size());
            serverPath = connectedServerList.get(serverIndex);
        }

        System.out.println("当前请求的服务器地址：" + serverPath);
        //发送消息
        long start = System.currentTimeMillis();
        Channel channel = channelMap.remove(serverPath);
        connectedServerList.remove(serverPath);
        rpcClientHandler.setChannel(channel);
        rpcClientHandler.setReqMsg(msg);

        Future future = executorService.submit(rpcClientHandler);
        Object result = future.get();
        long end = System.currentTimeMillis();
        System.out.println("当前请求耗时："+ (end - start)/1000 + "秒");
        channel.close();

        //记录请求信息
        Map<String,Object> nodeData = new HashMap<>();
        nodeData.put("ttlb",end -start);//请求耗时
        nodeData.put("lastReqTime",end);//最后一次请求响应时间
        nodeData.put("server",serverPath);
        if(zkClient.exists(nameSpace + serverPath)) {
            zkClient.writeData(nameSpace + serverPath, JSON.toJSONString(nodeData));
        }
        return result;
    }
}
